package com.example.demo.controller;
import com.example.demo.common.constant.CommonConstants;
import com.example.demo.common.response.ResponseData;
import com.example.demo.util.DateUtil;
import com.example.demo.util.common.ResponseDataUtil;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;

/**
 * (Test1)控制层
 *
 * @author wbx
 * @since 2021-05-14 11:16:12
 */
@RestController
@Api(value = "kafka发送消息", tags = "by wbx")
@RequestMapping("/kafka")
public class KafkaController {
    /**
     * 服务
     */
    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    @PostMapping("/sendMessage")
    @ApiOperation("kafka发送消息")
    public ResponseData<String> test1Add(){
        String message = "hello world kafka======>" + DateUtil.getCurrentDateStr();
        ListenableFuture<SendResult<String, String>> send = kafkaTemplate.send(CommonConstants.KAFKA_TOPIC, message);
        if(send.isCancelled()) {
            message = "消息发送失败";
        }
        return ResponseDataUtil.responseData(message);
    }
}
